package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.dto.ArticleStreamMessage;
import com.heima.article.dto.UpdateArticleMessage;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.util.StringUtils;

import java.time.Duration;

@EnableBinding(IHotArticleProcessor.class)
public class HotArticleListener {

    @StreamListener("hot_article_score_topic")
    @SendTo("hot_article_result_topic")
    public KStream<String, String> process(KStream<String, String> input) {
        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                // 收到的value 为  UpdateArticleMessage 对象
                // 将字符串json转换为对象  接收到的: {"add":1,"articleId":1460812407872970754,"type":1}
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                // 提取文章id,构建新的 KeyValue 键值对  key为文章id
                KeyValue<String, String> keyValue = new KeyValue<>(updateArticleMessage.getArticleId().toString(), value);
                return keyValue;
            }
        });
        // 根据文章id分组
        KGroupedStream<String, String> groupByKey = map.groupByKey();
        // 统计10秒内的每一篇文章所有的行为操作
        TimeWindowedKStream<String, String> windowedBy = groupByKey.windowedBy(TimeWindows.of(Duration.ofSeconds(10)));
        // 10秒内有多次消息,进行多次消息的结果聚合
        Initializer<String> init = new Initializer<String>() {
            @Override
            public String apply() {
                // 第一条消息进入后聚合的结果为空
                return null;
            }
        };
        Aggregator<String, String, String> agg = new Aggregator<String, String, String>() {
            @Override
            public String apply(String key, String value, String aggregate) {
                // String key, String value, String aggregate  key 指的是上面的keyValue对象中的key
                // value 指的是上面的keyValue对象中的value
                // aggregate 指的是上一次聚合的结果(10秒内的历史汇总数据)
                // 判断上一次的结果是否为空,不为空的话需要反序列化
                // 结果为  ArticleStreamMessage 对象
                ArticleStreamMessage result = null;
                if (StringUtils.isEmpty(aggregate)) {
                    result = new ArticleStreamMessage();
                    result.setArticleId(Long.parseLong(key));
                    result.setView(0);
                    result.setLike(0);
                    result.setComment(0);
                    result.setCollect(0);
                } else {
                    // 解析上一次的结果
                    result = JSON.parseObject(aggregate, ArticleStreamMessage.class);
                }
                // 根据当前的这条行为操作消息更新result 中的数据
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                switch (updateArticleMessage.getType()) {
                    case 0:
                        // 阅读
                        result.setView(result.getView() + updateArticleMessage.getAdd());
                        break;
                    case 1:
                        // 点赞
                        result.setLike(result.getLike() + updateArticleMessage.getAdd());
                        break;
                    case 2:
                        // 评论
                        result.setComment(result.getComment() + updateArticleMessage.getAdd());
                        break;
                    case 3:
                        // 收藏
                        result.setCollect(result.getCollect() + updateArticleMessage.getAdd());
                        break;
                }
                String json = JSON.toJSONString(result);
                return json;
            }
        };
        KTable<Windowed<String>, String> aggregate = windowedBy.aggregate(init, agg);
        // 生成结果,结果是每一篇文章10秒内所有的行为操作
        KStream<String, String> resultStream = aggregate.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue<>(key.key(), value);
            }
        });
        return resultStream;
    }
}
